package biz.ws.controller;

import api.common.result.Result;
import api.common.result.SystemException;
import biz.ws.ws.Message;
import biz.ws.ws.WsUtils;
import biz.ws.ws.client.WsClient;
import com.alibaba.fastjson.annotation.JSONField;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.util.concurrent.Promise;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import io.swagger.annotations.ApiOperation;
import lombok.Getter;
import lombok.Setter;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Consumer;

/**
 * @创建人 wsdc
 * @时间 2021/4/3
 * @描述 模拟压力测试
 * <li>    单独跑一个测试实例
 */
@RestController
@RequestMapping("/client/test")
@Api
public class ClientTestController {
    AtomicBoolean inUser = new AtomicBoolean(false);

    WsClient client = null;
    Semaphore sem = new Semaphore(0);

    @ApiOperation(value = "测试链接",tags = "Websocket客户端测试工具")
    @PostMapping("/connect")
    public Result connect(@RequestBody ConnectInfo info) throws Exception{
        checkInUse();

        Result rtn = new Result();

        client = new WsClient();
        client.connect(info.url, new Consumer<ChannelFuture>() {
            @Override
            public void accept(ChannelFuture future) {
                future.channel().close();
                if(future.isSuccess()){
                    rtn.data("success");
                    sem.release();
                    return ;
                }

                try{
                    Object o = ((Promise) future).get();
                    rtn.code(-1).data(o);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
                sem.release();
            }
        });

        sem.acquire();
        releaseInUse();
        client.close();
        client = null;

        return rtn;
    }

    @ApiOperation(value = "测试链接同时发送心跳",tags = "Websocket客户端测试工具")
    @PostMapping("/connectAndBeat")
    public Result connectAndBeat(@RequestBody ConnectInfo info) throws Exception{
        checkInUse();

        Result rtn = new Result();

        client = new WsClient(false, new WsClient.OnDataReceive() {
            @Override
            public void receiveData(ChannelHandlerContext ctx, WebSocketFrame webSocketFrame) {
                //  等待服务段传递过来的信息
                rtn.data(((TextWebSocketFrame) webSocketFrame).text());
                sem.release();
                ctx.channel().close();
            }
        });

        client.connect(info.url, new Consumer<ChannelFuture>() {
            @Override
            public void accept(ChannelFuture future) {
                if(future.isSuccess()){
                    WsUtils.write(future.channel(), Message.beat(1l));
                    return ;
                }

                //  没有连接上
                try{
                    Object o = ((Promise) future).get();
                    rtn.code(-1).data(((Exception) o).getMessage());
                    sem.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
        });

        sem.acquire();
        releaseInUse();
        return rtn;
    }

    @ApiOperation(value = "测试链接同时发送一条消息",tags = "Websocket客户端测试工具")
    @PostMapping("/connectAndSend")
    public Result connectAnSend(@RequestBody ConnectInfo info) throws Exception{
        checkInUse();

        Result rtn = new Result();

        client = new WsClient();

        client.connect(info.url, new Consumer<ChannelFuture>() {
            @Override
            public void accept(ChannelFuture future) {
                if(future.isSuccess()){
                    WsUtils.write(future.channel(), Message.sendTo(1l,2l,"forLove"));
                    return ;
                }

                //  没有连接上
                try{
                    Object o = ((Promise) future).get();
                    rtn.code(-1).data(((Exception) o).getMessage());
                    sem.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
        });

        LockSupport.parkNanos(5l * 1000 * 1000 * 1000);

        releaseInUse();
        return rtn;
    }

    ConcurrentConnectAndBeatResult r = null;

    @ApiOperation(value = "并发测试测试链接同时发送心跳",tags = "Websocket客户端测试工具")
    @PostMapping("/connectAndBeatConcurrent")
    public Result connectAndBeatConcurrent(@RequestBody ConcurrentConnectInfo info) throws Exception{
        checkInUse();
        Result rtn = new Result();
        r = new ConcurrentConnectAndBeatResult();
        r.count = info.count;
        rtn.setData(r);

        Map<Integer, LocalDateTime> beatStartTimeMap = new ConcurrentHashMap<>();
        client = new WsClient(false, new WsClient.OnDataReceive() {
            @Override
            public void receiveData(ChannelHandlerContext ctx, WebSocketFrame webSocketFrame) {
                //  等待服务段传递过来的信息
                LocalDateTime s = beatStartTimeMap.get(ctx.channel().hashCode());
                LocalDateTime e = LocalDateTime.now();
                int beatExpense = Duration.between(s, e).getNano() / 1000 / 1000;
                r.allBeatTime += beatExpense;
                r.maxBeatTime = r.maxBeatTime > beatExpense ? r.maxBeatTime : beatExpense;
                r.beatNum ++;
            }
        });

        for (int i = 0; i < info.count; i++) {
            LocalDateTime connectStartTime = LocalDateTime.now();
            client.connect(String.format(info.baseUrl,i), new Consumer<ChannelFuture>() {
                @Override
                public void accept(ChannelFuture future) {
                    LocalDateTime connectEndTime = LocalDateTime.now();
                    int connectExpense = Duration.between(connectStartTime, connectEndTime).getNano() / 1000 / 1000;
                    r.allConnectTime += connectExpense;
                    r.maxConnectTime = r.maxConnectTime > connectExpense ? r.maxConnectTime : connectExpense;
                    if(future.isSuccess()){
                        r.connectSuccessNum++;

                        //  随机时间发送报文
                        Mono.delay(Duration.ofSeconds(info.randomBeatTime))
                                .subscribe(k -> {
                                    //  记录心跳发送的时间
                                    beatStartTimeMap.put(future.channel().hashCode(),LocalDateTime.now());
                                    WsUtils.write(future.channel(), Message.beat(1l));
                                });

                        future.channel().closeFuture().addListener(new ChannelFutureListener() {
                            @Override
                            public void operationComplete(ChannelFuture future) throws Exception {
                                int k = 0;
                            }
                        });

                        return ;
                    }

                    //  没有连接上
                    Exception e0 = null;
                    try{
                        Object o = ((Promise) future).get();
                    } catch (Exception e) {
                       e0 = e;
                    } finally {
                        r.connectFailureNum++;
                        r.failureReasonList.add(e0.getMessage());
                    }
                }
            });
        }

        return rtn.msg("background working");
    }

    @GetMapping("/connectAndBeatConcurrent/result")
    @ApiOperation(value = "获取并发结果",tags = "Websocket客户端测试工具")
    public Result concurrentResult() throws Exception{
        if(r == null){
            return new Result();
        }

        LocalDateTime now = LocalDateTime.now();
        if(Duration.between(r.startTime,now).getSeconds() < r.randomBeatTime * 2){
            return new Result().data(r).msg("wording");
        }

        releaseInUse();

        //  数据统计
        r.averageBeatTime = r.allBeatTime / r.count;
        r.averageConnectTime = r.allConnectTime / r.count;

        client.close();
        client = null;

        Result data = new Result().data(r);
        r = null;
        return data;
    }

    public void checkInUse(){
        if(!inUser.compareAndSet(false,true)){
            throw new SystemException(-1,"inUse");
        }
        inUser.set(true);
    }

    public void releaseInUse(){
        inUser.set(false);
    }


    @Setter
    @Getter
    @ApiModel
    public static class ConnectInfo {
        @ApiModelProperty(value = "ws://192.168.31.99:7777/ws/ksv.do?userId=1&loginFrom=window")
        private String url;
        private Long userId;
    }

    @Setter
    @Getter
    @ApiModel
    public static class ConcurrentConnectInfo {
        @ApiModelProperty(value = "ws://192.168.31.99:7777/ws/ksv.do?userId=%d&loginFrom=window")
        private String baseUrl;
        private int count ;
        private int randomBeatTime;
    }

    @Setter
    @Getter
    public static class ConcurrentConnectAndBeatResult {
        private LocalDateTime startTime = LocalDateTime.now();
        private int count = 0;

        private int maxConnectTime = 0;
        private int averageConnectTime = 0;
        private int allConnectTime = 0;

        private int maxBeatTime = 0;
        private int averageBeatTime = 0;
        private int allBeatTime = 0;

        private int speedTime = 0;
        private int randomBeatTime = 0;

        private int connectSuccessNum = 0;
        private int connectFailureNum = 0;
        private int beatNum = 0;
        @JSONField(serialize = false)
        private List<String> failureReasonList = new LinkedList<>();
    }
}
